package com.pengheng.mqtt.platform;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 消息推送客户端
 */
@Slf4j
@Component
public class MqttPushClient {

    @Resource
    private MqttClient mqttClient;

    /**
     * 发布
     *
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public void publish(String topic, String pushMessage) {
        this.publish(0, false, topic, pushMessage);
    }

    /**
     * 发布
     *
     * @param qos         连接方式
     * @param retained    是否保留
     * @param topic       主题
     * @param pushMessage 消息体
     */
    public void publish(int qos, boolean retained, String topic, String pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        MqttTopic mTopic = mqttClient.getTopic(topic);
        if (null == mTopic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            log.error(e.getMessage(), e);
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void publish(String topic, byte[] pushMessage) {
        this.publish(0, false, topic, pushMessage);
    }

    public void publish(int qos, boolean retained, String topic, byte[] pushMessage) {
        MqttMessage message = new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage);
        MqttTopic mTopic = mqttClient.getTopic(topic);
        if (null == mTopic) {
            log.error("topic not exist");
        }
        MqttDeliveryToken token;
        try {
            token = mTopic.publish(message);
            token.waitForCompletion();
        } catch (MqttPersistenceException e) {
            log.error(e.getMessage(), e);
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题
     */
    public void subscribe(String topic) {
        this.subscribe(topic, 0);
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题
     * @param qos   连接方式
     */
    public void subscribe(String topic, int qos) {
        try {
            mqttClient.subscribe(topic, qos);
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 订阅某个主题
     *
     * @param topic 主题
     */
    public void subscribe(String[] topic) {
        log.info("开始订阅主题" + JSONUtil.parse(topic));
        try {
            mqttClient.subscribe(topic);
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }

    public void subscribe(String[] topic, int[] qos) {
        log.info("开始订阅主题" + JSONUtil.parse(topic));
        try {
            mqttClient.subscribe(topic, qos);
        } catch (MqttException e) {
            log.error(e.getMessage(), e);
        }
    }
}

